package com.alex.pull;

import com.alex.constants.Const;
import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.client.consumer.MQPullConsumerScheduleService;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 这个是不会因为服务再次启动, 而导致数据重新消费.
 */
public class PullSchedulePullConsumer {

    public static void main(String[] args) throws MQClientException {
        MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("test_schedule_pull_consumer");
        scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(Const.NAMESRV_ADDR);
        scheduleService.setMessageModel(MessageModel.CLUSTERING);

        // 指定从哪个Topic拉取消息
        scheduleService.registerPullTaskCallback("test-pull-topic", (mq, context) -> {
            MQPullConsumer consumer = context.getPullConsumer();
            System.out.println("queueId: " + mq.getQueueId() + ", ----");
            try {
                long offset = consumer.fetchConsumeOffset(mq, false);
                if (offset < 0) {
                    offset = 0;
                }

                // MessageQueue
                // Tag
                // Offset
                // 最大拉取条数
                PullResult pullResult = consumer.pull(mq, "*", offset, 32);
                switch (pullResult.getPullStatus()) {
                    case FOUND:
                        List<MessageExt> msgs = pullResult.getMsgFoundList();
                        for (MessageExt msg : msgs) {
                            System.out.println(new String(msg.getBody()));
                        }
                        break;
                    case OFFSET_ILLEGAL:
                    case NO_MATCHED_MSG:
                    case NO_NEW_MSG:
                        break;
                    default:
                        break;
                }

                // 更新offset位置
                consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());

                // 设置多少时间之后重新拉取消息(ms)
                context.setPullNextDelayTimeMillis(3000);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        });

        // 必须先注册, 再启动consumer
        scheduleService.start();
    }
}
